/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl.statistics.analyzer;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.api.domain.ExecutorStatistics;
import spring.cloud.tasks.api.domain.JobStatistics;
import spring.cloud.tasks.api.domain.NameSpaceConfiguration;
import spring.cloud.tasks.api.utils.ExecutorNodePath;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author timmy.hu
 */
public class ExecutorInfoAnalyzer {

    private static final Logger log = LoggerFactory.getLogger(ExecutorInfoAnalyzer.class);

    private final Map<String/** {executorName}-{domain} */, ExecutorStatistics> executorMap = new ConcurrentHashMap<>();


    private final AtomicInteger exeInDocker = new AtomicInteger(0);

    private final AtomicInteger exeNotInDocker = new AtomicInteger(0);

    public void analyzeExecutor(ZooKeeper zooKeeper,
                                NameSpaceConfiguration config) {
        long executorNumber = 0L; // 该域的在线executor数量
        // 统计物理容器资源，统计版本数据
        if (!zooKeeper.checkExists(ExecutorNodePath.getExecutorNodePath())) {

            return;
        }

        List<String> executors = zooKeeper.getChildren(ExecutorNodePath.getExecutorNodePath());
        if (executors == null) {

            return;
        }


        for (String exe : executors) {
            // 在线的才统计
            if (zooKeeper.checkExists(ExecutorNodePath.getExecutorIpNodePath(exe))) {
                // 统计是物理机还是容器
                String executorMapKey = exe + "-" + config.getNameSpace();
                ExecutorStatistics executorStatistics = executorMap.get(executorMapKey);
                if (executorStatistics == null) {
                    executorStatistics = new ExecutorStatistics(exe, config.getNameSpace());
                    executorStatistics.setNns(config.getNameSpace());
                    executorStatistics.setIp(zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(exe)));
                    executorMap.put(executorMapKey, executorStatistics);
                }
                // set runInDocker field
                if (isExecutorInDocker(zooKeeper, exe)) {
                    executorStatistics.setRunInDocker(true);
                    exeInDocker.incrementAndGet();
                } else {
                    exeNotInDocker.incrementAndGet();
                }
            }

        }
        executorNumber = executors.size();

    }

    public boolean isExecutorInDocker(ZooKeeper zooKeeper,
                                      String executorName) {
        return zooKeeper.checkExists(ExecutorNodePath.get$ExecutorTaskNodePath(executorName));
    }


    public void analyzeServer(ZooKeeper zooKeeper, List<String> servers, String job,
                              String nns,
                              NameSpaceConfiguration config, int loadLevel, JobStatistics jobStatistics) {
        for (String server : servers) {
            // 如果结点存活，算两样东西：1.遍历所有servers节点里面的processSuccessCount &
            // processFailureCount，用以统计作业每天的执行次数；2.统计executor的loadLevel;，
            if (!zooKeeper.checkExists(TaskNodePath.getExecutorStatus(job, server))) {
                continue;
            }
            // 1.遍历所有servers节点里面的processSuccessCount && processFailureCount，用以统计作业每天的执行次数；
            calcJobProcessCount(zooKeeper, nns, server, job, config.getNameSpace(), jobStatistics);

            // 2.统计executor的loadLevel
            calcLoadLevelOfExecutors(zooKeeper, nns, server, job, config.getNameSpace(), loadLevel, jobStatistics);
        }
    }

    private void calcJobProcessCount(ZooKeeper zooKeeper,
                                     String nns, String server, String job, String namespace,
                                     JobStatistics jobStatistics) {
        // 1.遍历所有servers节点里面的processSuccessCount &
        // processFailureCount，用以统计作业每天的执行次数；
        try {
            String processSuccessCountOfThisExeStr = zooKeeper
                    .getData(TaskNodePath.getProcessSucessCount(job, server));
            String processFailureCountOfThisExeStr = zooKeeper
                    .getData(TaskNodePath.getProcessFailureCount(job, server));
            long processSuccessCountOfThisExe = StringUtils.isBlank(processSuccessCountOfThisExeStr) ? 0
                    : Long.parseLong(processSuccessCountOfThisExeStr);
            long processFailureCountOfThisExe = StringUtils.isBlank(processFailureCountOfThisExeStr) ? 0
                    : Long.parseLong(processFailureCountOfThisExeStr);

            // executor当天运行成功失败数
            String executorMapKey = server + "-" + namespace;
            ExecutorStatistics executorStatistics = executorMap.get(executorMapKey);
            if (executorStatistics == null) {
                executorStatistics = new ExecutorStatistics(server, namespace);
                executorStatistics.setNns(nns);
                executorStatistics
                        .setIp(zooKeeper.getData(ExecutorNodePath.getExecutorIpNodePath(server)));
                executorMap.put(executorMapKey, executorStatistics);
            }
            executorStatistics.setFailureCountOfTheDay(
                    executorStatistics.getFailureCountOfTheDay() + processFailureCountOfThisExe);
            executorStatistics.setProcessCountOfTheDay(executorStatistics.getProcessCountOfTheDay()
                    + processSuccessCountOfThisExe + processFailureCountOfThisExe);
            jobStatistics.setProcessCountOfTheDay(processSuccessCountOfThisExe + processFailureCountOfThisExe);
            jobStatistics.setFailureCountOfTheDay(processFailureCountOfThisExe);
        } catch (Exception e) {
            log.info(e.getMessage(), e);
        }
    }

    private void calcLoadLevelOfExecutors(ZooKeeper zooKeeper,
                                          String nns,
                                          String server, String job, String namespace,
                                          int loadLevel, JobStatistics jobStatistics) {
        try {
            // enabled 的作业才需要计算权重
            if (!Boolean.parseBoolean(zooKeeper.getData(TaskNodePath.getConfigNodePath(job, "enabled")))) {
                return;
            }

            String sharding = zooKeeper.getData(TaskNodePath.getServerSharding(job, server));
            if (StringUtils.isNotEmpty(sharding)) {
                // 更新job的executorsAndshards
                String exesAndShards = (jobStatistics.getExecutorsAndShards() == null ? ""
                        : jobStatistics.getExecutorsAndShards()) + server + ":" + sharding + "; ";
                jobStatistics.setExecutorsAndShards(exesAndShards);
                // 2.统计是物理机还是容器
                String executorMapKey = server + "-" + namespace;
                ExecutorStatistics executorStatistics = executorMap.get(executorMapKey);
                if (executorStatistics == null) {
                    executorStatistics = new ExecutorStatistics(server, namespace);
                    executorStatistics.setNns(nns);
                    executorStatistics.setIp(zooKeeper
                            .getData(ExecutorNodePath.getExecutorIpNodePath(server)));
                    executorMap.put(executorMapKey, executorStatistics);
                    // set runInDocker field
                    if (isExecutorInDocker(zooKeeper, server)) {
                        executorStatistics.setRunInDocker(true);
                        exeInDocker.incrementAndGet();
                    } else {
                        exeNotInDocker.incrementAndGet();
                    }
                }
                if (executorStatistics.getJobAndShardings() != null) {
                    executorStatistics.setJobAndShardings(
                            executorStatistics.getJobAndShardings() + job + ":" + sharding + ";");
                } else {
                    executorStatistics.setJobAndShardings(job + ":" + sharding + ";");
                }
                int newLoad = executorStatistics.getLoadLevel() + (loadLevel * sharding.split(",").length);
                executorStatistics.setLoadLevel(newLoad);
            }
        } catch (Exception e) {
            log.info(e.getMessage(), e);
        }
    }

    public Map<String, ExecutorStatistics> getExecutorMap() {
        return executorMap;
    }

    public List<ExecutorStatistics> getExecutorList() {
        return new ArrayList<ExecutorStatistics>(executorMap.values());
    }


    public int getExeInDocker() {
        return exeInDocker.get();
    }

    public int getExeNotInDocker() {
        return exeNotInDocker.get();
    }
}
